package org.aceor.mddal.gateway.resultset;

import com.google.common.base.Objects;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import org.aceor.mddal.common.sqlparser.result.context.MergeContext;
import org.aceor.mddal.common.sqlparser.result.object.AggregationColumn;
import org.aceor.mddal.common.sqlparser.result.object.GroupByColumn;
import org.aceor.mddal.common.sqlparser.result.object.OrderByColumn;
import org.aceor.mddal.common.executor.ExecuteUnit;
import org.aceor.mddal.gateway.executor.ExecutorEngineFactory;
import org.aceor.mddal.common.executor.MergeUnit;
import org.aceor.mddal.gateway.merger.common.ResultSetQueryIndex;
import org.aceor.mddal.gateway.merger.function.AggregateFunction;
import org.aceor.mddal.gateway.merger.groupby.GroupByKey;
import org.aceor.mddal.gateway.merger.groupby.GroupByValue;
import org.aceor.mddal.gateway.util.AggregateFunctionUtil;
import org.aceor.mddal.gateway.util.ResultSetUtil;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.*;

/**
 * Created by lxue on 16/4/16.
 */
public class GroupByResultSet extends BaseResultSet implements ResultSet {

    private static final Logger log = LoggerFactory.getLogger(GroupByResultSet.class);

    private final List<GroupByColumn> groupByColumns;

    private final List<OrderByColumn> orderByColumns;

    private final List<AggregationColumn> aggregationColumns;

    private final ResultSetMetaData resultSetMetaData;

    private Iterator<GroupByValue> groupByResultIterator;

    private GroupByValue currentGroupByResultSet;

    public GroupByResultSet(List<ResultSet> resultSets, MergeContext mergeContext) throws SQLException {
        super(resultSets, mergeContext.getLimit());
        aggregationColumns = mergeContext.getAggregationColumns();
        groupByColumns = mergeContext.getGroupByColumns();
        orderByColumns = mergeContext.getOrderByColumns();
        resultSetMetaData = getResultSets().iterator().next().getMetaData();
    }

    @Override
    protected boolean nextRowSet() throws SQLException {
        if (null == groupByResultIterator) {
            groupByResultIterator = reduce(map()).iterator();
        }
        if (groupByResultIterator.hasNext()) {
            currentGroupByResultSet = groupByResultIterator.next();
            return true;
        } else {
            return false;
        }
    }

    private Multimap<GroupByKey, GroupByValue> map() throws SQLException {
        ExecuteUnit<ResultSet, Map<GroupByKey, GroupByValue>> executeUnit = new ExecuteUnit<ResultSet, Map<GroupByKey, GroupByValue>>() {
            @Override
            public Map<GroupByKey, GroupByValue> execute(ResultSet input) throws Exception {
                Map<GroupByKey, GroupByValue> map = Maps.newHashMap();
                /**
                 * 把每个groupby组里面的行记录解释成GroupByValue,groupBy字段组成GroupByKey
                 */
                while (input.next()) {
                    GroupByValue groupByValue = new GroupByValue();
                    for (int count = 1; count <= resultSetMetaData.getColumnCount(); count++) {
                        groupByValue.put(count, resultSetMetaData.getColumnLabel(count), input.getObject(count));
                    }
                    GroupByKey groupByKey = new GroupByKey();
                    for (GroupByColumn each : groupByColumns) {
                        groupByKey.append(ResultSetUtil.getValue(each.getName(), each.getAlias(), input).toString());
                    }
                    map.put(groupByKey, groupByValue);
                }
                return map;
            }
        };
        /**
         * 多个ResultSet中Map按key合并(其实并没有合并,用了MultiMap)
         */
        MergeUnit<Map<GroupByKey, GroupByValue>, Multimap<GroupByKey, GroupByValue>> mergeUnit = new MergeUnit<Map<GroupByKey, GroupByValue>, Multimap<GroupByKey, GroupByValue>>() {
            @Override
            public Multimap<GroupByKey, GroupByValue> merge(List<Map<GroupByKey, GroupByValue>> params) {
                Multimap<GroupByKey, GroupByValue> multimap = HashMultimap.create();
                for (Map<GroupByKey, GroupByValue> map : params) {
                    for (Map.Entry<GroupByKey, GroupByValue> entry : map.entrySet()) {
                        multimap.put(entry.getKey(), entry.getValue());
                    }
                }
                return multimap;
            }
        };
        return ExecutorEngineFactory
                .getInstance()
                .execute(getResultSets(), executeUnit, mergeUnit);
    }

    /**
     * Multimap中合并相同key的所有GroupByValue
     *
     * @param mappedResult
     * @return
     * @throws SQLException
     */
    private Collection<GroupByValue> reduce(final Multimap<GroupByKey, GroupByValue> mappedResult) throws SQLException {
        List<GroupByValue> groupByValueList = Lists.newArrayList();
        for (GroupByKey groupByKey : mappedResult.keySet()) {
            Collection<GroupByValue> groupByValues = mappedResult.get(groupByKey);
            GroupByValue mergedGroupByValue = new GroupByValue();
            {
                /**
                 * [核心]记录合并
                 * 首先明确,groupby下select字段可以有groupby字段和聚集函数字段
                 * 其次明确,两种情况最终都只会合并成一条记录.该记录的特点是:groupby字段照搬(不同分库分表下的rs都一个值),聚集函数字段就是要通过求值,得到最终结果
                 * 因此,求groupByValue这行记录,就只要分别求上述两种类型的值即可
                 */
                //for循环把这一行记录的每列值都算出来
                for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
                    String columnName = resultSetMetaData.getColumnLabel(i);
                    AggregationColumn aggregationColumn = findAggregationColumn(columnName);
                    Object value;
                    if (null != aggregationColumn) {
                        //聚集函数计算合并的值
                        //暂时支持单一参数 通过aggregationColumn.getArguments().get(0)获得ColumnName
                        value = aggregate(aggregationColumn, new ResultSetQueryIndex(columnName), groupByValues);
                    } else {
                        //就是groupby字段值.每个groupByValues里面都一样,取第一个计算即可
                        value = groupByValues.iterator().next().getValue(new ResultSetQueryIndex(i));
                    }
                    mergedGroupByValue.put(i, columnName, value);
                }
            }
            /**
             * 用于排序
             */
            mergedGroupByValue.setGroupByColumnList(groupByColumns);
            mergedGroupByValue.setOrderByColumnList(orderByColumns);
            groupByValueList.add(mergedGroupByValue);
        }
        Collections.sort(groupByValueList);
        return groupByValueList;
    }

    private Object aggregate(AggregationColumn aggregationColumn, ResultSetQueryIndex queryIndex, Collection<GroupByValue> groupByValues) {
        AggregateFunction function = AggregateFunctionUtil
                .createAggregateFunction(
                        aggregationColumn.getAggregationType(),
                        groupByValues.iterator().next().getValue(queryIndex));
        for (GroupByValue groupByValue : groupByValues) {
            Object value = groupByValue.getValue(queryIndex);
            function.merge(value);
        }
        return function.getResult();
    }

    private AggregationColumn findAggregationColumn(String columnName) {
        if (CollectionUtils.isEmpty(aggregationColumns)) {
            return null;
        }
        for (AggregationColumn aggregationColumn : aggregationColumns) {
            if (Objects.equal(columnName, aggregationColumn.getExpression())) {
                return aggregationColumn;
            }
        }
        return null;
    }

    public GroupByValue getCurrentGroupByResultSet() {
        return currentGroupByResultSet;
    }
}
